import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

public class T2WorkQueues {
    final static String QUEUE_NAME = "task_queue";
    final static ExecutorService es = Executors.newFixedThreadPool(3);

    public static void main(String[] args) {
        es.submit(T2WorkQueues::send);
        es.submit(T2WorkQueues::recv);
        es.submit(T2WorkQueues::recv2);
    }

    static void send() {
        try (Connection connection = ConnectionUtil.getConnection();
             Channel channel = connection.createChannel()) {
            // 创建队列
            // 参数 1：队列名称
            // 参数 2：是否定义持久化队列
            // 参数 3：是否独占本次连接
            // 参数 4：是否在不使用的时候自动删除队列
            // 参数 5：队列其它参数
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            String message = "Hello World!";
            while (true) {
                Thread.sleep(1000);
                // 发送消息
                // 参数 1：交换机名称，如果没有指定则使用默认 Default Exchage
                // 参数 2：路由 key，简单模式可以传递队列名称
                // 参数 3：消息属性：PERSISTENT_TEXT_PLAIN(将消息标记为持久性)
                // 参数 4：消息内容
                channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
                channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
                System.out.println(Thread.currentThread().getName() + " [x] Sent '" + message + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    static void recv() {
        try {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.basicQos(1); // 公平调度，一次只接受一条未确认的消息
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println(Thread.currentThread().getName() + " [x] Received '" + message + "'");
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // 消息确认并要求服务器将该消息从队列中删除
                // channel.basicReject(delivery.getEnvelope().getDeliveryTag(), true); // 拒绝消息并将其丢弃
                // channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); // 拒绝消息并将其重新放回队列中
            };
            // 订阅消息
            // 爹数 1：队列名称
            // 参数 2：是否自动确认。设置为 true 表示消息按收到后自动向 mq 回复接收到了，mq 接收到回复后会删除消息：设置为 false 则需要手动确认
            channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
            });
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }

    static void recv2() {
        try {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.basicQos(1); // 公平调度，一次只接受一条未确认的消息
            Consumer consumer = new DefaultConsumer(channel) {
                /**
                 * @param consumerTag 消息者标签，在 channel.basicConsumel 时候可以指定
                 * @param envelope    消息包内容，可从中获取消息 id，消息 routingkey，交换机，消息和重转标记(收到消息失败后是否需要重新发送)
                 * @param properties  消息属性
                 * @param body        消息
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("ThreadName：" + Thread.currentThread().getName());
                    System.out.println("路由 key：" + envelope.getRoutingKey());
                    System.out.println("交换机：" + envelope.getExchange());
                    System.out.println("消息 id：" + envelope.getDeliveryTag());
                    System.out.println("消息：" + new String(body, StandardCharsets.UTF_8));
                }
            };
            // 订阅消息，一条消息只会有一个消费者
            // 爹数 1：队列名称
            // 参数 2：是否自动确认。设置为 true 表示消息按收到后自动向 mq 回复接收到了，mq 接收到回复后会删除消息：设置为 false 则需要手动确认
            channel.basicConsume(QUEUE_NAME, true, consumer);
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}